package com.hzmg.akkacloud.worker.common.utils;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import com.google.common.collect.Maps;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.model.WorkerInfo;
import com.hzmg.akkacloud.common.serialize.SerializerUtils;
import com.hzmg.akkacloud.worker.background.ReactiveDiscoveryClient;
import com.hzmg.akkacloud.worker.pojo.RouterInfo;
import com.hzmg.akkacloud.worker.starter.AkkaStarter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * 远程调用工具
 * 这部分代码感觉写得不是特别优雅
 *
 * @author zbw
 * @since 2022/8/14
 */
public class RemoteInvokeUtils {
    private static final Logger log = LoggerFactory.getLogger(RemoteInvokeUtils.class);
    //应该不会有线程问题吧，反正router创建出来后会自动更新，且actor重复会报错的。
    private static final Map<String, ActorRef> routers = Maps.newConcurrentMap();
    private static ActorSystem actorSystem = AkkaStarter.actorSystem;

    /**
     * 远程服务调用（异步调用非阻塞），携带返回值
     *
     * @param <T>              返回参数类型
     * @param parameters       参数
     * @param actorName        所需调用方法的注解名称
     * @param remoteServerName 远程服务名
     * @return
     */
    public static <T> T getRemoteServiceResult(Object parameters, String actorName, String remoteServerName) {
        ActorRef router = routers.get(actorName);
        byte[] bytes = SerializerUtils.serialize(parameters);
        if (router == null) {
            //为了避免临时增加服务，额外增加了两倍；这边的处理方式以后会改动，比如修改成可在后台手动配置。
            int length = ReactiveDiscoveryClient.getAllWorkersWithAppName(remoteServerName).size() * 2;
            ActorSelection actorSelection = actorSystem.actorSelection("/user/" + RemoteConstant.ROUTER_CREATER_ACTOR_NAME);
            try {
                //return AkkaUtils.handleWhenAsk(actorSelection, new RouterInfo(actorName, Collections.singletonList(actorName), length), SerializerUtils.deSerialized(AkkaUtils.remoteAskByRouter(routers.get(actorName), bytes)));
                return AkkaUtils.handleWhenAsk(actorSelection, new RouterInfo(actorName, Collections.singletonList(remoteServerName), length),
                        () -> AkkaUtils.remoteAskByRouter(routers.get(actorName), bytes));
            } catch (Exception e) {
                throw new AkkaCloudException(e);
            }
        } else {
            //return SerializerUtils.deSerialized(AkkaUtils.remoteAskByRouter(router, bytes));
            return (T) AkkaUtils.remoteAskByRouter(router, bytes);
        }
    }

    /**
     * 远程服务调用，携带返回值，指定角色。
     *
     * @param <T>              返回参数类型
     * @param parameters       参数
     * @param actorName        所需调用方法的注解名称
     * @param remoteServerName 远程服务名
     * @param roles            调用指定角色
     * @return
     */
    public static <T> Object getRemoteServiceResultWithRoles(Object parameters, String actorName, String remoteServerName, List<String> roles) {
        ActorRef router = routers.get(actorName);
        byte[] bytes = SerializerUtils.serialize(parameters);
        if (router == null) {
            //为了避免临时增加服务，额外增加了两倍；这边的处理方式以后会改动，比如修改成可在后台手动配置。
            int length = ReactiveDiscoveryClient.getAllWorkersWithAppName(remoteServerName).size() * 2;
            ActorSelection actorSelection = actorSystem.actorSelection("/user/" + RemoteConstant.ROUTER_CREATER_ACTOR_NAME);
            try {
                return AkkaUtils.handleWhenAsk(actorSelection, new RouterInfo(actorName, roles, length), () -> AkkaUtils.remoteAskByRouter(routers.get(actorName), bytes));
            } catch (Exception e) {
                throw new AkkaCloudException(e);
            }
        } else {
            return AkkaUtils.remoteAskByRouter(router, bytes);
        }
    }

    /**
     * 远程服务调用，不携带返回值。
     *
     * @param parameters
     * @param actorName
     * @param remoteServerName
     */
    public static void getRemoteService(Object parameters, String actorName, String remoteServerName) {
        ActorRef router = routers.get(actorName);
        List<String> roles = Collections.singletonList(remoteServerName);
        byte[] bytes = SerializerUtils.serialize(parameters);
        runInNewRouter(actorName, remoteServerName, roles, router, bytes);

    }

    /**
     * 远程服务调用，不携带返回值，指定角色。
     *
     * @param parameters
     * @param actorName
     * @param remoteServerName
     * @param roles
     */
    public static void getRemoteServiceRoles(Object parameters, String actorName, String remoteServerName, List<String> roles) {
        ActorRef router = routers.get(actorName);
        byte[] bytes = SerializerUtils.serialize(parameters);
        runInNewRouter(actorName, remoteServerName, roles, router, bytes);
    }

    private static void runInNewRouter(String actorName, String remoteServerName, List<String> roles, ActorRef router, byte[] bytes) {
        if (router == null) {
            //为了允许临时增加服务，额外增加了两倍空间存储路由；
            // 这边的处理方式以后会改动，计划修改成可在后台管理界面手动配置。
            int length = ReactiveDiscoveryClient.getAllWorkersWithAppName(remoteServerName).size() * 2;
            ActorSelection actorSelection = actorSystem.actorSelection("/user/" + RemoteConstant.ROUTER_CREATER_ACTOR_NAME);
            try {
                AkkaUtils.handleWhenAskIgnore(actorSelection, new RouterInfo(actorName, roles, length), () -> AkkaUtils.remoteAskByRouterIgnore(routers.get(actorName), bytes));
            } catch (Exception e) {
                throw new AkkaCloudException(e);
            }
        } else {
            AkkaUtils.remoteAskByRouterIgnore(router, bytes);
        }
    }

    private static WorkerInfo randomInstance(List<WorkerInfo> workerInfoList) {
        long randomNum = System.currentTimeMillis();
        return workerInfoList.get((int) (randomNum % workerInfoList.size()));
    }

    public static Map<String, ActorRef> getRouters() {
        return routers;
    }

}

